package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo03Burks {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo03Burks")
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    // 加载数据
    val burkDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("burk String,year String,tsl01 INT,tsl02 INT,tsl03 INT,tsl04 INT,tsl05 INT,tsl06 INT,tsl07 INT,tsl08 INT,tsl09 INT,tsl10 INT,tsl11 INT,tsl12 INT")
      .load("Spark/data/burk/burks.txt")

    burkDF.cache()

    //    burkDF.show()

    // 将每行数据转换成12行数据，即每个公司每个月一条数据
    // 将DF注册成临时视图
    burkDF.createOrReplaceTempView("burks")

    // 1、统计每个公司每年按月累计收入
    spark
      .sql(
        """
          |select  t1.burk
          |        ,t1.year
          |        ,t1.month
          |        ,t1.tsl
          |        ,sum(tsl) over (partition by burk,year order by month) as sum_tsl
          |from (
          |    select  burk
          |            ,year
          |            ,month+1 as month
          |            ,tsl
          |    from burks lateral view posexplode(array(tsl01,tsl02,tsl03,tsl04,tsl05,tsl06,tsl07,tsl08,tsl09,tsl10,tsl11,tsl12)) v1 as month,tsl
          |) t1
          |""".stripMargin)
      .show(20)

    burkDF
      .select($"burk", $"year", posexplode(array($"tsl01", $"tsl02", $"tsl03", $"tsl04", $"tsl05", $"tsl06", $"tsl07", $"tsl08", $"tsl09", $"tsl10", $"tsl11", $"tsl12")).as(Array("month", "tsl")))
      .select($"burk", $"year", $"month" + 1 as "month", $"tsl")
      .withColumn("sum_tsl", sum($"tsl") over Window.partitionBy($"burk", $"year").orderBy($"month"))
      .show(20)

    // 2、统计每个公司当月比上年同期增长率（同比）
    spark
      .sql(
        """
          |select  tt1.burk
          |        ,tt1.month
          |        ,tt1.year
          |        ,tt1.tsl
          |        ,tt1.last_tsl
          |        ,round((tt1.tsl-tt1.last_tsl)/tt1.last_tsl,3) as tb
          |from (
          |    select  t1.burk
          |            ,t1.year
          |            ,t1.month
          |            ,t1.tsl
          |            ,lag(tsl) over (partition by burk,month order by year) as last_tsl
          |    from (
          |        select  burk
          |                ,year
          |                ,month+1 as month
          |                ,tsl
          |        from burks lateral view posexplode(array(tsl01,tsl02,tsl03,tsl04,tsl05,tsl06,tsl07,tsl08,tsl09,tsl10,tsl11,tsl12)) v1 as month,tsl
          |    ) t1
          |) tt1
          |""".stripMargin)
      .show(100)

    burkDF
      .select($"burk", $"year", posexplode(array($"tsl01", $"tsl02", $"tsl03", $"tsl04", $"tsl05", $"tsl06", $"tsl07", $"tsl08", $"tsl09", $"tsl10", $"tsl11", $"tsl12")).as(Array("month", "tsl")))
      .select($"burk", $"year", $"month" + 1 as "month", $"tsl")
      .withColumn("last_tsl", lag($"tsl", 1) over Window.partitionBy($"burk", $"month").orderBy($"year"))
      .withColumn("tb", round(($"tsl" - $"last_tsl") * 100 / $"last_tsl", 3))
      .show()

  }

}
